package xyj.sparrow.scheduler.cron.stream;

import cn.hutool.core.net.NetUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.util.Assert;
import xyj.sparrow.scheduler.cron.Cron;
import xyj.sparrow.scheduler.cron.ops.OpsMsg;
import xyj.sparrow.scheduler.cron.thread.MonitorThread;
import xyj.sparrow.scheduler.entity.App;
import xyj.sparrow.scheduler.entity.Task;
import xyj.sparrow.scheduler.mapper.AppMapper;
import xyj.sparrow.scheduler.mapper.TaskMapper;
import xyj.sparrow.scheduler.server.ClientManager;
import xyj.sparrow.scheduler.util.ApplicationContextUtil;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

/**
 * cron消息消费者
 *
 * @author 江南小俊
 * @since 2021/6/20 下午11:49
 **/
@Slf4j
public class CronConsumer {
    private final ExecutorService executorService = Executors.newCachedThreadPool();
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private LettuceConnectionFactory redisConnectionFactory;
    @Resource
    private MonitorThread monitorThread;

    @Value("${xyj.sparrow.server.port}")
    private int port;

    @PostConstruct
    private void init() {
        log.info("初始消息消费者，开启消息监听");
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, OpsMsg>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                        .batchSize(10)
                        .executor(executorService)
                        .pollTimeout(Duration.ZERO)
                        .targetType(OpsMsg.class)
                        .build();
        StreamMessageListenerContainer<String, ObjectRecord<String, OpsMsg>> container =
                StreamMessageListenerContainer.create(redisConnectionFactory, options);

        prepareChannelAndGroup(stringRedisTemplate.opsForStream(), "channel:stream:cron", "group:cron");
        container.receiveAutoAck(Consumer.from("group:cron", "c_" + NetUtil.getLocalhostStr() + ":" + port),
                StreamOffset.create("channel:stream:cron", ReadOffset.lastConsumed()),
                new StreamMessageListener(monitorThread));

        container.start();
    }

    private void prepareChannelAndGroup(StreamOperations<String, Object, Object> ops, String stream,
                                        String group) {
        String status = "OK";
        try {
            StreamInfo.XInfoGroups groups = ops.groups(stream);
            if (groups.stream().noneMatch(xInfoGroup -> group.equals(xInfoGroup.groupName()))) {
                status = ops.createGroup(stream, group);
            }
        } catch (Exception exception) {
            RecordId initialRecord = ops.add(ObjectRecord.create(stream, "Initial Record"));
            Assert.notNull(initialRecord, "Cannot initialize stream with key '" + stream + "'");
            status = ops.createGroup(stream, ReadOffset.from(initialRecord), group);
        } finally {
            Assert.isTrue("OK".equals(status), "Cannot create group with name '" + group + "'");
        }
    }

    public static class StreamMessageListener implements StreamListener<String, ObjectRecord<String, OpsMsg>> {
        private MonitorThread monitorThread;

        public StreamMessageListener(MonitorThread monitorThread) {
            this.monitorThread = monitorThread;
        }

        @Override
        public void onMessage(ObjectRecord<String, OpsMsg> message) {
            log.info("收到消息：{}", message);
            OpsMsg opsMsg = message.getValue();
            switch (opsMsg.getOpsType()) {
                case OpsMsg.OpsType.REMOVE_TASK:
                    monitorThread.removeByTaskIds(opsMsg.getIds());
                    break;
                case OpsMsg.OpsType.REMOVE_APP:
                    monitorThread.removeByAppIds(opsMsg.getIds());
                    break;
                case OpsMsg.OpsType.ADD_CRON: {
                    List<Cron> cronList = listTask(opsMsg.getIds());
                    if (cronList != null && !cronList.isEmpty()) {
                        for (Cron cron : cronList) {
                            monitorThread.addCron(cron);
                        }
                    }
                }
                break;
                case OpsMsg.OpsType.UPDATE_CRON: {
                    List<Cron> cronList = listTask(opsMsg.getIds());
                    if (cronList != null && !cronList.isEmpty()) {
                        for (Cron cron : cronList) {
                            monitorThread.updateCron(cron);
                        }
                    }
                }
                break;
                case OpsMsg.OpsType.UPDATE_CRON_APP: {
                    List<Cron> cronList = listTaskByAppId(opsMsg.getIds());
                    if (cronList != null && !cronList.isEmpty()) {
                        for (Cron cron : cronList) {
                            monitorThread.updateCron(cron);
                        }
                    }
                }
                break;
                case OpsMsg.OpsType.UPDATE_CRON_TASK: {
                    List<Cron> cronList = listTaskByTaskId(opsMsg.getIds());
                    if (cronList != null && !cronList.isEmpty()) {
                        for (Cron cron : cronList) {
                            monitorThread.updateCron(cron);
                        }
                    }
                }
                break;
                case OpsMsg.OpsType.RUN_CRON_TASK: {
                    TaskMapper taskMapper = ApplicationContextUtil.getBean(TaskMapper.class);
                    Task task = taskMapper.selectById(opsMsg.getIds().get(0));
                    if (task != null) {
                        ClientManager.scheduling(task.getAppId(), task.getTaskId(), LocalDateTime.now());
                    }
                }
                break;
            }
        }

        private List<Cron> listTaskByTaskId(List<String> taskIds) {
            TaskMapper taskMapper = ApplicationContextUtil.getBean(TaskMapper.class);
            QueryWrapper<Task> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("task_status", "1");
            queryWrapper.in("task_id", taskIds);
            List<Task> tasks = taskMapper.selectList(queryWrapper);
            return tasks.stream().map(task -> new Cron(monitorThread, task)).collect(Collectors.toList());
        }

        private List<Cron> listTaskByAppId(List<String> appIds) {
            TaskMapper taskMapper = ApplicationContextUtil.getBean(TaskMapper.class);
            List<String> enabledAppIds = listEnabledAppIds();
            QueryWrapper<Task> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("task_status", "1");
            queryWrapper.in("app_id", appIds);
            List<Task> tasks = taskMapper.selectList(queryWrapper);
            tasks = tasks.stream().filter(task -> enabledAppIds.contains(task.getAppId())).collect(Collectors.toList());
            return tasks.stream().map(task -> new Cron(monitorThread, task)).collect(Collectors.toList());
        }

        private List<Cron> listTask(List<String> taskIds) {
            List<String> enabledAppIds = listEnabledAppIds();
            TaskMapper taskMapper = ApplicationContextUtil.getBean(TaskMapper.class);
            QueryWrapper<Task> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("task_status", "1");
            queryWrapper.in("task_id", taskIds);
            List<Task> tasks = taskMapper.selectList(queryWrapper);
            tasks = tasks.stream().filter(task -> enabledAppIds.contains(task.getAppId())).collect(Collectors.toList());
            return tasks.stream().map(task -> new Cron(monitorThread, task)).collect(Collectors.toList());
        }

        private List<String> listEnabledAppIds() {
            AppMapper appMapper = ApplicationContextUtil.getBean(AppMapper.class);
            QueryWrapper<App> appQueryWrapper = new QueryWrapper<>();
            appQueryWrapper.eq("app_status", 1);
            List<App> apps = appMapper.selectList(appQueryWrapper);
            return apps.stream().map(App::getAppId).collect(Collectors.toList());
        }
    }


}
